package com.ruyuan2020.im.gateway.route.strategy.consistent;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.HashUtil;
import com.ruyuan2020.im.common.core.exception.SystemException;

import java.util.Collection;
import java.util.Iterator;
import java.util.SortedMap;
import java.util.TreeMap;

/**
 * 一致性hash路由
 *
 * @author zhonghuashishan
 */
public class ConsistentHashRoute<T extends Node> {

    /**
     * Tree实现一个虚拟节点环形结构
     * 默认是有序的，会根据key的大小去进行排序，按照key hash值来进行排序
     */
    private final SortedMap<Long, VirtualNode<T>> ring = new TreeMap<>();

    /**
     * 哈希函数
     */
    private final HashFunction hashFunction;

    public ConsistentHashRoute(Collection<T> physicalNodes, int virtualNodeCount) {
        this(physicalNodes, virtualNodeCount, new Murmur32Hash());
    }

    /**
     * @param physicalNodes    物理节点集合
     * @param virtualNodeCount 虚拟节点数量
     * @param hashFunction     哈希函数
     */
    public ConsistentHashRoute(Collection<T> physicalNodes, int virtualNodeCount, HashFunction hashFunction) {
        if (hashFunction == null) {
            throw new SystemException("哈希函数不能为空");
        }
        this.hashFunction = hashFunction;
        // 如果物理节点不为空
        // 物理节点就是我们的路由服务的各个机器的节点
        if (CollUtil.isNotEmpty(physicalNodes)) {
            // 遍历物理节点，把物理节点添加到虚拟的哈希环中，128个虚拟节点
            for (T physicalNode : physicalNodes) {
                addNode(physicalNode, virtualNodeCount);
            }
        }
    }

    /**
     * 添加物理节点到虚拟环中
     * 一致性hash策略，搞一个环，每个物理节点都有一堆虚拟节点，如果说没有虚拟节点的话
     * 可能很多数据一开始都是hash路由到某个物理节点，这个物理节点宕机崩溃了以后
     * 这批数据可能都会路由到另外的一个物理节点去，会导致热点问题，某个物理节点压力很大
     * 虚拟节点，万一要是某个物理节点崩溃了，所有的路由请求都会分散到各个其他节点的虚拟节点上去
     * 尽可能的分散一下路由压力，给各个节点来均摊
     *
     * @param physicalNode     物理节点
     * @param virtualNodeCount the number of virtual node of the physical node. Value should be greater than or equals to 0
     */
    public void addNode(T physicalNode, int virtualNodeCount) {
        if (virtualNodeCount < 0) throw new SystemException("虚拟节点数不能小于0");
        // 获取物理节点副本数
        long replicas = getReplicas(physicalNode);
        for (int i = 0; i < virtualNodeCount; i++) {
            VirtualNode<T> virtualNode = new VirtualNode<>(physicalNode, i + replicas);
            ring.put(hashFunction.hash(virtualNode.getKey()), virtualNode);
        }
    }

    /**
     * 从环中删除物理节点
     */
    public void removeNode(T physicalNode) {
        Iterator<Long> it = ring.keySet().iterator();
        while (it.hasNext()) {
            Long key = it.next();
            VirtualNode<T> virtualNode = ring.get(key);
            if (virtualNode.isVirtualNodeOf(physicalNode)) {
                it.remove();
            }
        }
    }

    public T routeNode(String objectKey) {
        // key是群id
        if (ring.isEmpty()) {
            return null;
        }
        Long hashVal = hashFunction.hash(objectKey);
        // 群id的hash值，在ring里所有的虚拟节点hash key但是大于等于群id hash值的话，拿出来，作为一个子map
        SortedMap<Long, VirtualNode<T>> tailMap = ring.tailMap(hashVal);
        // 子map是非空，拿到这个子map里第一个节点，如果子map是空，就直接从ring环里，拿到第一个节点
        Long nodeHashVal = !tailMap.isEmpty() ? tailMap.firstKey() : ring.firstKey();

        // 物理节点没有宕机，物理节点很少，有一种可能会导致热点问题，不均匀，很多群id路由到了某一个物理节点上去
        // 每个物理节点加了128个虚拟节点之后，尽可能让不同的群id均匀的分摊路由到不同的虚拟节点上去
        // 通过虚拟节点再路由到物理节点上去，尽可能让群id均匀分摊到不同的物理节点上去
        return ring.get(nodeHashVal).getPhysicalNode();
    }


    /**
     * 获取已存在环中的物理节点副本数
     * 针对一个物理节点获取他在环里已经有的虚拟节点的数量，副本数量
     */
    public long getReplicas(T physicalNode) {
        return ring.values().stream().filter(virtualNode -> virtualNode.isVirtualNodeOf(physicalNode)).count();
    }


    /**
     * 默认实现的哈希函数
     */
    private static class Murmur32Hash implements HashFunction {

        @Override
        public long hash(String key) {
            return HashUtil.murmur32(key.getBytes());
        }
    }

}
